package com.zzl.spark.parquet;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

public class JDBCDataSource {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("JDBCDataSource");
        JavaSparkContext sc = new JavaSparkContext(conf);
        System.setProperty("hadoop.home.dir", "E:\\hadoop");

        SQLContext sqlContext = new SQLContext(sc);
        //方法1
//        Map<String,String> options = new HashMap<>();
//        options.put("url","jdbc:mysql://zhangzeli-node1:3306/testdb");
//        options.put("driver","com.mysql.jdbc.Driver");
//        options.put("user","spark");
//        options.put("password","spark");
//        options.put("dbtable","student_info");
//        Dataset<Row> jdbc = sqlContext.read().format("jdbc").options(options).load();
//        jdbc.show();
        DataFrameReader reader = sqlContext.read().format("jdbc");
        reader.option("url","jdbc:mysql://192.168.137.10:3306/testdb?useSSL=false");
        reader.option("driver","com.mysql.jdbc.Driver");
        reader.option("user","root");
        reader.option("password","123456");
        reader.option("dbtable","student_info");
        Dataset<Row> studentinfo = reader.load();
        reader.option("dbtable","student_scores");
        Dataset<Row> studentscores = reader.load();
        studentinfo.createOrReplaceTempView("studentInfo");
        studentscores.createOrReplaceTempView("studentScores");

        String sql ="select i.*,s.score  from studentInfo i join studentScores s on i.name=s.name  where s.score>80";
        Dataset<Row> result = sqlContext.sql(sql);
        result.show();

        result.javaRDD().foreachPartition((f)->{
            Class.forName("com.mysql.jdbc.Driver");
            Connection root = DriverManager.getConnection("jdbc:mysql://192.168.137.10:3306/testdb?useSSL=false", "root", "123456");
            PreparedStatement p = root.prepareStatement("insert into good_student_info values(?,?,?)");
            f.forEachRemaining(data->{
                try {
                    p.setString(1,data.getAs("name"));
                    p.setInt(2,data.getAs("age"));
                    p.setInt(3,data.getAs("score"));
                    p.executeUpdate();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            });
            p.close();
            root.close();
        });

        sc.stop();
    }
}
